package fun.liuyu2783.txmessage.trade.mq;

import fun.liuyu2783.txmessage.trade.dto.Trade;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.HashMap;
import java.util.UUID;

@Slf4j
@RestController
@RequestMapping("message")
public class MqProviderController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private static final String ORDER_TOPIC = "ORDER_TOPIC";
    private static final String ORDER_TOPIC_ORDERLY = "ORDER_TOPIC_ORDERLY";

    /**
     * 普通消息
     */
    @PostMapping("send/{tag}")
    public HashMap send(@PathVariable(value = "tag") String tag){
        Trade trade = new Trade();
        String orderId = UUID.randomUUID().toString();
        trade.setId(orderId);
        trade.setGoodsId("1");
        trade.setGoodsName("可口可乐");
        trade.setName("liuyu2783");
        trade.setCreateTime(new Date());
        rocketMQTemplate.send(ORDER_TOPIC +":"+tag, MessageBuilder.withPayload(trade).build());
        HashMap map = new HashMap();
        map.put("result", "消息发送成功");
        return map;
    }

    /**
     * 同步消息
     */
    @PostMapping("syncSend/{tag}")
    public HashMap syncSend(@PathVariable(value = "tag") String tag){
        Trade trade = new Trade();
        String orderId = UUID.randomUUID().toString();
        trade.setId(orderId);
        trade.setGoodsId("1");
        trade.setGoodsName("可口可乐");
        trade.setName("liuyu2783");
        trade.setCreateTime(new Date());
        SendResult sendResult = rocketMQTemplate.syncSend(ORDER_TOPIC +":"+tag, MessageBuilder.withPayload(trade).build());
        log.info("同步消息返回值，result={}", sendResult);
        HashMap map = new HashMap();
        map.put("result", sendResult);
        return map;
    }

    /**
     * 异步消息
     */
    @PostMapping("asyncSend/{tag}")
    public HashMap asyncSend(@PathVariable(value = "tag") String tag){
        Trade trade = new Trade();
        String orderId = UUID.randomUUID().toString();
        trade.setId(orderId);
        trade.setGoodsId("1");
        trade.setGoodsName("可口可乐");
        trade.setName("liuyu2783");
        trade.setCreateTime(new Date());
        rocketMQTemplate.asyncSend(ORDER_TOPIC + ":"+tag, MessageBuilder.withPayload(trade).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步消息返回值，result={}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("异步消息异常，throwable={}", throwable);
            }
        });
        HashMap map = new HashMap();
        map.put("result", "异步消息发送成功");
        return map;
    }

    /**
     * 发送延时消息（上面的发送同步消息，delayLevel的值就为0，因为不延时）
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    @PostMapping("sendDelay/{tag}/{delayLevel}")
    public HashMap sendDelayMsg(@PathVariable(value = "tag") String tag, @PathVariable(value = "delayLevel") int delayLevel) {
        Trade trade = new Trade();
        String orderId = UUID.randomUUID().toString();
        trade.setId(orderId);
        trade.setGoodsId("1");
        trade.setGoodsName("可口可乐");
        trade.setName("liuyu2783");
        trade.setCreateTime(new Date());
        SendResult sendResult = rocketMQTemplate.syncSend(ORDER_TOPIC + ":" + tag, MessageBuilder.withPayload(trade).build(), 3000, delayLevel);
        HashMap map = new HashMap();
        map.put("result", sendResult);
        return map;
    }

    /**
     * 异步多条消息
     */
    @PostMapping("asyncSendMsgs/{tag}")
    public HashMap asyncSendMsgs(@PathVariable(value = "tag") String tag){
        for (int i = 0; i < 10; i++) {
            Trade trade = new Trade();
            String orderId = UUID.randomUUID().toString();
            trade.setId(orderId);
            trade.setGoodsId("1");
            trade.setGoodsName("可口可乐");
            trade.setName("liuyu2783");
            trade.setCreateTime(new Date());
            rocketMQTemplate.asyncSend(ORDER_TOPIC + ":"+tag, MessageBuilder.withPayload(trade).build(), new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步消息返回值，result={}", sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("异步消息异常，throwable={}", throwable);
                }
            });
        }
        HashMap map = new HashMap();
        map.put("result", "异步消息发送成功");
        return map;
    }

    /**
     * 顺序消息
     */
    @PostMapping("orderly/{tag}")
    public HashMap orderly(@PathVariable(value = "tag") String tag) throws InterruptedException {
        String orderId = UUID.randomUUID().toString();
        for (int i = 0; i < 10; i++) {
            Trade trade = new Trade();
            trade.setId(orderId+"-"+i);
            trade.setGoodsId("1");
            trade.setGoodsName("可口可乐");
            trade.setName("liuyu2783");
            trade.setCreateTime(new Date());
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(ORDER_TOPIC_ORDERLY + ":" + tag, MessageBuilder.withPayload(trade).build(), orderId);
            log.info("顺序消息发送，message={}, status={}", trade, sendResult.getSendStatus());
        }
        HashMap map = new HashMap();
        map.put("result", "异步消息发送成功");
        return map;
    }

}
